﻿using DataSync.Entity;
using log4net;
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using Boco.CommonLib;
using DataSync.Common;
using System.Xml.Linq;
using Newtonsoft.Json;

namespace DataSync.Proc.Writer
{
    class DBWriter : DataSyncWriterBase
    {
        private static ILog log = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

        protected DBOutputConfig CurOutputDbConf;    //目的数据库配置
        protected IDbConnection ConnIns; //当前数据库连接
        protected readonly string CNT_Sql_Ins = "Sql_Ins";
        protected readonly string CNT_Sql_Upt = "Sql_Upt";
        protected readonly string CNT_Sql_Pre = "Sql_Pre";
        protected readonly string CNT_Sql_Post = "Sql_Post";
        protected readonly string CNT_Sql_CheckExists = "Sql_CheckExists";
        protected readonly int CNT_MaxErrorNum = 30;

        protected override void ReadConfig(XElement xmlnode)
        {
            //读取配置
            var dbconf = xmlnode.Element("DBConfig");
            if (dbconf == null)
            {
                throw new ApplicationException("DBConfig配置不正确");
            }
            DBOutputConfig conf = new DBOutputConfig()
            {
                DBType = (DataBaseType)Enum.Parse(typeof(DataBaseType), dbconf.GetAttribute("DBType"), false),
                ConnString = dbconf.GetAttribute("ConnString"),
                ProcType = dbconf.GetAttribute("ProcType"),
                TableName = dbconf.GetAttribute("TableName"),
                KeyCol = new List<string>(),
                Sqls = new Dictionary<string, string>(),
                Type = "DB"
            };

            string KeyCols = dbconf.GetAttribute("KeyCol");
            if (!string.IsNullOrEmpty(KeyCols))
            {
                var cols = KeyCols.Split(new string[] { ";", "|" }, StringSplitOptions.RemoveEmptyEntries);
                conf.KeyCol = cols.ToList();
            }
            var sqls = xmlnode.Element("Sqls").Elements("Sql"); //.SelectNodes("Sqls/Sql");
            foreach (var item in sqls)
            {
                var id = item.GetAttribute("id");
                if (conf.Sqls.ContainsKey(id))
                {
                    conf.Sqls[id] = item.Value;
                }
                else
                {
                    conf.Sqls.Add(id, item.Value);
                }
            }
            //for (int i = 0; i < sqls.Count; i++)
            //{
            //    var id = sqls[i].GetAttribute("id");
            //    if (conf.Sqls.ContainsKey(id))
            //    {
            //        conf.Sqls[id] = sqls[i].InnerText;
            //    }
            //    else
            //    {
            //        conf.Sqls.Add(id, sqls[i].InnerText);
            //    }
            //}

            //检查正确性
            if (string.IsNullOrEmpty(conf.ConnString))
            {
                throw new ApplicationException("DBConfig:ConnString配置不正确");
            }
            if (string.IsNullOrEmpty(conf.ProcType))
            {
                conf.ProcType = "Add";//默认为增量采集
            }
            if (string.IsNullOrEmpty(conf.Sqls[CNT_Sql_Ins]))
            {
                throw new ApplicationException("Sqls:Sql_Ins配置不正确");
            }
            CurOutputDbConf = conf;
        }

        public override int Execute(IList<Dictionary<string, ColData>> data)
        {
            log.DebugFormat("入库数据开始：{0}条", data.Count);
            var procnum = 0;
            var conn = GetCurConn();

            if (CurOutputDbConf.ProcType.ToUpper() == "SYNC")
            {
                DBSync(data, conn);//同步模式
            }
            else
            {
                DBInsert(data, conn);
            }

            procnum = data.Count;
            log.DebugFormat("入库数据结束：{0}条", data.Count);
            return procnum;
        }

        public override void PostExecute()
        {
            //执行后
            if (CurOutputDbConf.Sqls.ContainsKey(CNT_Sql_Post) && !string.IsNullOrEmpty(CurOutputDbConf.Sqls[CNT_Sql_Post]))
            {
                var sqls = CurOutputDbConf.Sqls[CNT_Sql_Post].Split(';');
                using (var conn = GetCurConn())
                using (var trs = conn.BeginTransaction())  //开启事务
                {
                    foreach (var sql in sqls)
                    {
                        log.InfoFormat("执行结束Sql_Post:[{0}]", sql);

                        var cmd = conn.CreateCommand();
                        cmd.CommandText = sql;
                        cmd.Transaction = trs;
                        cmd.ExecuteNonQuery();
                    }
                }
            }
        }

        public override void PreExecute()
        {
            //执行前
            if (CurOutputDbConf.Sqls.ContainsKey(CNT_Sql_Pre) && !string.IsNullOrEmpty(CurOutputDbConf.Sqls[CNT_Sql_Pre]))
            {
                var sqls = CurOutputDbConf.Sqls[CNT_Sql_Pre].Split(';');
                using (var conn = GetCurConn())
                using (var trs = conn.BeginTransaction())  //开启事务
                {
                    foreach (var sql in sqls)
                    {
                        log.InfoFormat("执行初始化Sql_Pre:[{0}]", sql);

                        var cmd = conn.CreateCommand();
                        cmd.Transaction = trs;
                        cmd.CommandText = sql;
                        cmd.ExecuteNonQuery();
                    }
                    trs.Commit();
                }
            }
        }

        protected IDbConnection GetCurConn()
        {
            if (ConnIns == null || string.IsNullOrEmpty(ConnIns.ConnectionString))
            {
                ConnIns = DBFactory.GetConn(this.CurOutputDbConf);
            }
            if (ConnIns.State == ConnectionState.Closed || ConnIns.State == ConnectionState.Broken)
            {
                ConnIns.Open();
            }
            return ConnIns;
        }

        public override void Close()
        {
            if (ConnIns != null)
            {
                if (ConnIns.State == ConnectionState.Open)
                {
                    ConnIns.Close();
                }
                ConnIns = null;
            }
            base.Close();
        }

        /// <summary>
        /// insert入库
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="data"></param>
        protected virtual void DBInsert(IList<Dictionary<string, ColData>> data, IDbConnection conn)
        {
            string sql = CurOutputDbConf.Sqls[CNT_Sql_Ins]; //入库sql
            using (var trs = conn.BeginTransaction())  //开启事务
            {
                var cmd = conn.CreateCommand();
                cmd.CommandText = sql;
                cmd.Transaction = trs;
                foreach (var row in data)
                {
                    cmd.Parameters.Clear();

                    foreach (var col in row)
                    {
                        var p = cmd.CreateParameter();
                        p.ParameterName = col.Key;
                        p.DbType = DBFactory.GetDbType(col.Value.FieldType);
                        p.Value = DataTypeHelper.ConvertToType(col.Value.FieldType, col.Value.Value);
                        p.Direction = ParameterDirection.Input;
                        cmd.Parameters.Add(p);
                    }
                    cmd.ExecuteNonQuery();
                }

                trs.Commit();
            }
        }

        /// <summary>
        /// 数据同步
        /// </summary>
        /// <param name="dt"></param>
        /// <param name="conn"></param>
        protected virtual void DBSync(IList<Dictionary<string, ColData>> data, IDbConnection conn)
        {
            string sql_ins = CurOutputDbConf.Sqls[CNT_Sql_Ins]; //新增sql
            string sql_upt = CurOutputDbConf.Sqls[CNT_Sql_Upt]; //更新sql
            string sql_exists = CurOutputDbConf.Sqls[CNT_Sql_CheckExists]; //检查是否存在sql

            int errorNum = 0;
            using (var trs = conn.BeginTransaction())  //开启事务
            {
                var cmd = conn.CreateCommand();
                cmd.Transaction = trs;
                foreach (var row in data)
                {
                    cmd.Parameters.Clear();
                    //检查数据是否存在
                    cmd.CommandText = sql_exists;
                    foreach (var kitem in this.CurOutputDbConf.KeyCol)
                    {
                        var co = row[kitem];
                        var p = cmd.CreateParameter();
                        p.ParameterName = kitem;
                        p.DbType = DBFactory.GetDbType(co.FieldType);
                        p.Value = co.Value;
                        p.Direction = ParameterDirection.Input;
                        cmd.Parameters.Add(p);
                    }
                    var rt = Convert.ToInt32(cmd.ExecuteScalar());

                    cmd.Parameters.Clear();
                    foreach (var col in row)
                    {
                        var p = cmd.CreateParameter();
                        p.ParameterName = col.Key;
                        p.DbType = DBFactory.GetDbType(col.Value.FieldType);
                        p.Value = DataTypeHelper.ConvertToType(col.Value.FieldType, col.Value.Value);
                        p.Direction = ParameterDirection.Input;
                        cmd.Parameters.Add(p);
                    }

                    if (rt > 0)
                    {
                        //数据存在
                        cmd.CommandText = sql_upt;
                        //调整参数顺序
                        foreach (var kitem in this.CurOutputDbConf.KeyCol)
                        {
                            cmd.Parameters.RemoveAt(kitem);
                            var co = row[kitem];
                            var p = cmd.CreateParameter();
                            p.ParameterName = kitem;
                            p.DbType = DBFactory.GetDbType(co.FieldType);
                            p.Value = co.Value;
                            p.Direction = ParameterDirection.Input;
                            cmd.Parameters.Add(p);
                        }
                    }
                    else
                    {
                        //数据不存在
                        cmd.CommandText = sql_ins;
                    }

                    try
                    {
                        var num= cmd.ExecuteNonQuery();
                    }
                    catch (Exception ex)
                    {
                        errorNum++;
                        log.Error(string.Format("数据{0}入库出错", JsonConvert.SerializeObject(row)), ex);
                    }
                }

                trs.Commit();
            }
        }
    }
}
